home *** CD-ROM | disk | FTP | other *** search
/ Cream of the Crop 26 / Cream of the Crop 26.iso / os2 / pvm34b3.zip / pvm34b3 / pvm3 / src / mppmsg.c < prev    next >
C/C++ Source or Header  |  1997-07-22  |  21KB  |  903 lines
  1.  
  2. static char rcsid[] =
  3.     "$Id: mppmsg.c,v 1.6 1997/06/25 22:09:13 pvmsrc Exp $";
  4.  
  5. /*
  6.  *         PVM version 3.4:  Parallel Virtual Machine System
  7.  *               University of Tennessee, Knoxville TN.
  8.  *           Oak Ridge National Laboratory, Oak Ridge TN.
  9.  *                   Emory University, Atlanta GA.
  10.  *      Authors:  J. J. Dongarra, G. E. Fagg, M. Fischer
  11.  *          G. A. Geist, J. A. Kohl, R. J. Manchek, P. Mucci,
  12.  *         P. M. Papadopoulos, S. L. Scott, and V. S. Sunderam
  13.  *                   (C) 1997 All Rights Reserved
  14.  *
  15.  *                              NOTICE
  16.  *
  17.  * Permission to use, copy, modify, and distribute this software and
  18.  * its documentation for any purpose and without fee is hereby granted
  19.  * provided that the above copyright notice appear in all copies and
  20.  * that both the copyright notice and this permission notice appear in
  21.  * supporting documentation.
  22.  *
  23.  * Neither the Institutions (Emory University, Oak Ridge National
  24.  * Laboratory, and University of Tennessee) nor the Authors make any
  25.  * representations about the suitability of this software for any
  26.  * purpose.  This software is provided ``as is'' without express or
  27.  * implied warranty.
  28.  *
  29.  * PVM version 3 was funded in part by the U.S. Department of Energy,
  30.  * the National Science Foundation and the State of Tennessee.
  31.  */
  32.  
  33. /* Define low level message functions for the MPP port */
  34.  
  35. #include <stdio.h>
  36. #include <stdlib.h>
  37. #include <sys/types.h>
  38. #if defined(NEEDSSELECTH)
  39. #include <sys/select.h>
  40. #endif
  41. #include <errno.h>
  42. #include <sys/stat.h>
  43. #include <sys/socket.h>
  44. #ifndef NOUNIXDOM
  45. #include <sys/un.h>
  46. #endif
  47. #include <netinet/in.h>
  48. #include <netinet/tcp.h>
  49. #include <pvm3.h>
  50. #include <pvmproto.h>
  51. #include "lpvm.h"
  52. #include "global.h"
  53. #include "pvmalloc.h"
  54. #include "listmac.h"
  55. #include "lmsg.h"
  56. #include "mppmsg.h"
  57. #include "mppchunk.h"
  58. #include "pvmmimd.h"
  59. #include "bfunc.h"
  60.  
  61.  
  62. /* --- external declarations */
  63.  
  64. void hex_inadport __ProtoGlarp__ (( char *, struct sockaddr_in * ));
  65.  
  66. extern int errno;
  67. extern int pvmdebmask;           /* from pvmd.c */
  68.  
  69. #if defined(IMA_NODE)
  70. extern int pvmhostnode;
  71. #endif
  72.  
  73. #if defined(IMA_PGON)
  74. #include <nx.h>
  75.  
  76. #define ASYNCSEND(_app, _tag, _buf, _len, _dest, _partid, _mid)  \
  77.         _isend((long)(_tag), _buf, (long)(_len), (long)(_dest), (long)(_partid))
  78.  
  79. #define ASYNCRECV(_app, _src, _tag, _buf, _len, _partid, _info, _mid) \
  80.         _irecvx((long)(_tag), _buf, (long)(_len), (long)(_src), \
  81.                 (long)(_partid), (long *)_info)
  82.  
  83. #define MSGDONE(_app,_mid,_flag,_status) \
  84.         _msgdone(*(_mid))
  85.  
  86. #define MSGMERGE(_mid1, _mid2) \
  87.         _msgmerge(*(_mid1), *(_mid2))
  88.  
  89. #define MSGTAG(_info) \
  90.         _info[0]
  91.  
  92. #define MSGLEN(_info, _len) \
  93.         _info[1]
  94.  
  95. #define MSGSRC(_info) \
  96.         _info[2]
  97.  
  98. /* Host (daemon) uses the same nx routines that nodes do */
  99.  
  100. #define HOSTASYNCSEND ASYNCSEND
  101. #define HOSTASYNCRECV ASYNCRECV
  102. #define HOSTMSGMERGE MSGMERGE
  103. #define HOSTMSGDONE MSGDONE
  104. #define HOSTMSGTAG MSGTAG
  105. #define HOSTMSGLEN MSGLEN
  106. #define HOSTMSGSRC MSGSRC
  107.  
  108.  
  109. #endif /* IMA_PGON */
  110.  
  111. #if defined(IMA_SP2MPI)
  112. #if defined(IMA_NODE)
  113. #include <mpi.h>
  114.  
  115. #define ASYNCSEND(_app, _tag, _buf, _len, _dest, _partid, _mid) \
  116.         MPI_Isend((void *) (_buf), len, MPI_BYTE, _dest,\
  117.                  _tag, MPI_COMM_WORLD, _mid)
  118.  
  119. #define ASYNCRECV(_app, _src, _tag, _buf, _len, _partid, _info, _mid) \
  120.         MPI_Irecv((void *) (_buf), _len, MPI_BYTE, _src, _tag,\
  121.                 MPI_COMM_WORLD, _mid)
  122.  
  123. #define MSGDONE(_app,_mid,_flag,_status) \
  124.         MPI_Test(_mid, _flag, _status )
  125.  
  126. #define MSGTAG(_info) \
  127.         (_info)->MPI_TAG
  128.  
  129. #define MSGMERGE(_mid1, _mid2) NULL 
  130.         
  131.  
  132. #define MSGLEN(_info, _len) \
  133.         MPI_Get_count(_info, MPI_BYTE, _len) 
  134.  
  135. #define MSGSRC(_info) \
  136.         (_info)->MPI_SOURCE
  137.  
  138. #else /* IMA_NODE */
  139.  
  140. #define ASYNCSEND(_app, _tag, _buf, _len, _dest, _partid, _mid) 0
  141.  
  142. #define ASYNCRECV(_app, _src, _tag, _buf, _len, _partid, _info, _mid) 0
  143.  
  144. #define MSGDONE(_app,_mid,_flag,_status) 0 
  145.  
  146. #define MSGTAG(_info) 0 
  147.  
  148. #define MSGMERGE(_mid1, _mid2) NULL 
  149.         
  150. #define MSGLEN(_info, _len) 0 
  151.  
  152. #define MSGSRC(_info) 0 
  153.  
  154. #endif /*IMA_NODE*/
  155.  
  156. /* Relay process uses the same  routines that nodes do */
  157.  
  158. #define HOSTASYNCSEND ASYNCSEND
  159. #define HOSTASYNCRECV ASYNCRECV
  160. #define HOSTMSGMERGE MSGMERGE
  161. #define HOSTMSGDONE MSGDONE
  162. #define HOSTMSGTAG MSGTAG
  163. #define HOSTMSGLEN MSGLEN
  164. #define HOSTMSGSRC MSGSRC
  165.  
  166.  
  167. #endif /* IMA_SP2MPI */
  168.  
  169.  
  170. /* ----------( Node -- Node ) Routines ----------- */
  171. static int
  172. pvm_inodesend(appid, tag, buffer, len, dest, partid, mid)
  173. int appid;
  174. int tag;
  175. char *buffer;
  176. int len;
  177. int dest;
  178. int partid;
  179. msgmid_t *mid;
  180. {
  181. #if defined(IMA_PGON)
  182.     return ((*mid) = ASYNCSEND(appid, tag, buffer, len, dest, partid, mid));
  183. #endif
  184. #if defined(IMA_SP2MPI)
  185.     return  ASYNCSEND(appid, tag, buffer, len, dest, partid, mid);
  186. #endif
  187. }
  188.  
  189.  
  190. static int
  191. pvm_inoderecv(appid, src, tag, buffer, len, partid, info, mid)
  192. int appid;
  193. int src;
  194. int tag;
  195. char *buffer;
  196. int len;
  197. int partid;
  198. int *info;
  199. msgmid_t *mid;
  200. {
  201. #if defined(IMA_PGON)
  202.     return ((*mid)=ASYNCRECV(appid, src, tag, buffer, len, partid, info, mid));
  203. #endif
  204. #if defined(IMA_SP2MPI)
  205.     return ASYNCRECV(appid, src, tag, buffer, len, partid, info, mid);
  206. #endif
  207. }
  208.  
  209.  
  210. static msgmid_t 
  211. pvm_nodemsgmerge (mid1, mid2)
  212. msgmid_t *mid1;
  213. msgmid_t *mid2;
  214. {
  215.     return (msgmid_t) MSGMERGE(mid1, mid2);
  216. }
  217.  
  218. static int
  219. pvm_nodemsgdone (appid, mid, info)
  220. int appid;
  221. msgmid_t *mid;
  222. info_t *info;
  223. {
  224. int flag;
  225. #if defined(IMA_PGON)
  226.     return MSGDONE(appid, mid, &flag, info);
  227. #endif
  228. #if defined(IMA_SP2MPI)
  229.     MSGDONE(appid, mid, &flag, info);
  230.     return flag;
  231. #endif
  232. }
  233.  
  234. static int
  235. pvm_nodemsgsrc( info )
  236. info_t *info;
  237. {
  238.     return MSGSRC(info);
  239. }
  240.  
  241. static int
  242. pvm_nodemsgtag( info )
  243. info_t *info;
  244. {
  245.     return MSGTAG(info);
  246. }
  247.  
  248. static int
  249. pvm_nodemsglen( info )
  250. info_t *info;
  251. {
  252. int len;
  253. #if defined(IMA_PGON)
  254.     return MSGLEN(info, &len);
  255. #endif
  256. #if defined(IMA_SP2MPI)
  257.     MSGLEN(info,&len);
  258.     return len;
  259. #endif
  260. }
  261.  
  262. /* ---------- ( Host <--> Node ) Routines ----------- */
  263. static int
  264. pvm_ihostsend(appid, tag, buffer, len, dest, partid, mid)
  265. int appid;
  266. int tag;
  267. char *buffer;
  268. int len;
  269. int dest;
  270. int partid;
  271. msgmid_t *mid;
  272. {
  273. #if defined(IMA_PGON)
  274.     return ((*mid) = HOSTASYNCSEND(appid, tag, buffer, len, dest, partid, mid));
  275. #endif
  276. #if defined(IMA_SP2MPI)
  277.     return  HOSTASYNCSEND(appid, tag, buffer, len, dest, partid, mid);
  278. #endif
  279. }
  280.  
  281.  
  282. static int
  283. pvm_ihostrecv(appid, src, tag, buffer, len, partid, info, mid)
  284. int appid;
  285. int src;
  286. int tag;
  287. char *buffer;
  288. int len;
  289. int partid;
  290. int *info;
  291. msgmid_t *mid;
  292. {
  293. #if defined(IMA_PGON)
  294.     return ((*mid)=HOSTASYNCRECV(appid, src, tag, buffer, len, partid, info, mid));
  295. #endif
  296. #if defined(IMA_SP2MPI)
  297.     return HOSTASYNCRECV(appid, src, tag, buffer, len, partid, info, mid);
  298. #endif
  299. }
  300.  
  301. static msgmid_t 
  302. pvm_hostmsgmerge (mid1, mid2)
  303. msgmid_t *mid1;
  304. msgmid_t *mid2;
  305. {
  306.     return HOSTMSGMERGE(mid1, mid2);
  307. }
  308.  
  309. static int
  310. pvm_hostmsgdone (appid, mid, info)
  311. int appid;
  312. msgmid_t *mid;
  313. info_t *info;
  314. {
  315. int flag;
  316. #if defined(IMA_PGON)
  317.     return HOSTMSGDONE(appid, mid, &flag, info);
  318. #endif
  319. #if defined(IMA_SP2MPI)
  320.     HOSTMSGDONE(appid, mid, &flag, info);
  321.     return flag;
  322. #endif
  323. }
  324.  
  325. static int
  326. pvm_hostmsgsrc( info )
  327. info_t *info;
  328. {
  329.     return HOSTMSGSRC(info);
  330. }
  331.  
  332. static int
  333. pvm_hostmsgtag( info )
  334. info_t *info;
  335. {
  336.     return HOSTMSGTAG(info);
  337. }
  338.  
  339. static int
  340. pvm_hostmsglen( info )
  341. info_t *info;
  342. {
  343. int len;
  344. #if defined(IMA_PGON)
  345.     return HOSTMSGLEN(info, &len);
  346. #endif
  347. #if defined(IMA_SP2MPI)
  348.     HOSTMSGLEN(info,&len);
  349.     return len;
  350. #endif
  351. }
  352.  
  353.  
  354. static MSGFUNC nodemsgfunc = { pvm_inodesend,
  355.                         pvm_inoderecv,
  356. #if defined(IMA_PGON)
  357.                         pvm_nodemsgmerge,
  358. #else
  359.                         NULL,
  360. #endif
  361.                         pvm_nodemsgdone,
  362.                         pvm_nodemsglen,
  363.                         pvm_nodemsgsrc,
  364.                         pvm_nodemsgtag };
  365.  
  366. static MSGFUNC hostmsgfunc = { pvm_ihostsend,
  367.                         pvm_ihostrecv,
  368. #if defined(IMA_PGON)
  369.                         pvm_hostmsgmerge,
  370. #else
  371.                         NULL,
  372. #endif
  373.                         pvm_hostmsgdone,
  374.                         pvm_hostmsglen,
  375.                         pvm_hostmsgsrc,
  376.                         pvm_hostmsgtag };
  377.  
  378.  
  379. /* -------- External Routines --------- */
  380. MSGFUNC_PTR 
  381. pvm_nodemsgfunc()
  382. {
  383.     return &nodemsgfunc;
  384. } 
  385.  
  386. MSGFUNC_PTR
  387. pvm_hostmsgfunc()
  388. {
  389.     return &hostmsgfunc;
  390. }
  391.  
  392.  
  393. /* Define some intialization functions that are only used by node procs
  394. */
  395. #if defined(IMA_NODE)
  396.  
  397. /* ------ messsage system initialization, destruction ----- */
  398. int
  399. pvm_mpp_message_init(node, partsize, host, partid)
  400. int *node;
  401. int *partsize;
  402. int *host; 
  403. int *partid;
  404. {
  405.     char errtxt[64];
  406.     char *msgbuf;
  407.     char *p;
  408.  
  409.     int ac = 0;
  410.     int msgbufsiz;
  411.  
  412.     *node = -1;            /* initialize to bogus values */
  413.     *partsize = -1;
  414.     *host = -1;
  415.     *partid = 0;
  416.  
  417. #if defined(IMA_PGON)
  418.  
  419.     if ( (*partid = _myptype()) == INVALID_PTYPE ) 
  420.     {
  421.         pvmlogerror("mpp_message_init() no process type\n");
  422.         return PvmSysErr;
  423.     }
  424.  
  425.     _setptype(0);        /* always set ptype to 0 */
  426.  
  427.     *node = _mynode();
  428.     *partsize = _numnodes();
  429.     *host = *partsize;
  430. #endif
  431.  
  432. #if defined(IMA_SP2MPI)
  433.     MPI_Init(&ac, NULL);
  434.     MPI_Comm_rank(MPI_COMM_WORLD, node);
  435.     MPI_Comm_size(MPI_COMM_WORLD, partsize);
  436.     *host = *partsize - 1;        /* host is last process in the group */
  437.  
  438. /*    if (!(p = getenv("PVMBUFSIZE")) || !(msgbufsiz = strtol(p, (char**)0, 0)))
  439.         msgbufsiz = MPIBUFSIZ;
  440.     if (!(msgbuf = malloc(msgbufsiz)))
  441.         pvmlogerror("relay() out of memory");
  442.     MPI_Buffer_attach(msgbuf, msgbufsiz);           used in psend or relay */
  443.  
  444.         if (*node == *host)    /* I'm the host, run the host proc */
  445.     {
  446.             (void)pvmhost();
  447.     }
  448.     else
  449.         
  450.  
  451. /*    MPI_Bcast(pvminfo, SIZEHINFO, MPI_INT, pvmhostnode, MPI_COMM_WORLD); */
  452. #endif
  453.  
  454.     return 0;
  455. }
  456.  
  457. int
  458. pvm_mpp_message_stop() 
  459. {
  460. #if defined(IMA_SP2MPI)
  461.     MPI_Finalize();
  462. #endif
  463.     return 0;
  464. }
  465.     
  466. #endif /* IMA_NODE */
  467. /* ===============  Host and Relay Processes for MPI =============== */
  468. #if defined(IMA_SP2MPI) && defined(IMA_NODE)
  469.  
  470. #define NMPPSBUFMIDS 32            /* number of allowed outstanding send mids */
  471. static msgmid_t mppsendmids[NMPPSBUFMIDS];
  472. static CHUNK_PTR *mppoutchunks[NMPPSBUFMIDS];
  473. static int lastindex = 0;
  474.  
  475. /* ------------ relay -------------- */
  476. /* Relay messages between pvmd and node tasks. */
  477. void relay(dsock, numnodes)
  478.     int dsock;                    /* pvmd socket */
  479.     int numnodes;                /* number of nodes in the partition */
  480. {
  481.     fd_set wrk_rfds, wrk_wfds, rfds, wfds;
  482.     int nfds;
  483.     struct timeval tout;
  484.  
  485.     struct frag *topvmdq = 0;    /* frags to pvmd */
  486.     struct frag *frtask = 0;    /* (big) frag from task */
  487.     struct frag *totaskq;        /* frag being sent to task */
  488.     struct frag *fp;
  489.     struct frag *frpvmd = (struct frag *) NULL;
  490.     char *txcp = 0;             /* point to remainder of topvmd */
  491.     int txtogo = 0;             /* len of remainder of topvmd */
  492.     int toread;                    /* number of bytes to be read from pvmd */
  493.     int frtogo;                    /* len of remainder of a fragment */
  494.     int topvmd_dst;                /* dst of fragment being sent to pvmd */ 
  495.     int topvmd_src;                /* src of fragment being sent to pvmd */
  496.     int len;
  497.     int topvmd_ff;                /* ff of fragment being sent to pvmd */
  498.     int    dst;                    /* dst of fragment being sent to node */
  499.     int    node;                    /* node number */
  500.     MPI_Request rmhd;            /* msg IDs returned by async recv */
  501.     int n;
  502.     char *cp;
  503.     int err,i;
  504.     MPI_Status sta;                /* info on pending message */
  505.     int dummy;
  506.     int flag;                    /* MPI_Test result */
  507.     struct frag *hdr;
  508.     int taskbuf = 0;            /* current buffer number to probe */ 
  509.  
  510.  
  511.  
  512.  
  513.     MPP_DIRECTI_PTR taskdirect = (MPP_DIRECTI_PTR) NULL; /* ordering structs */
  514.  
  515.     MSG_INFO_PTR taskfrags = (MSG_INFO_PTR) NULL;
  516.  
  517.     CHUNK_PTR readyFrags = (CHUNK_PTR) NULL;
  518.  
  519.     MSGFUNC_PTR mfunc;
  520.     
  521.     struct frag *cftd = (struct frag *) NULL; /* cur.frag going to daemon */
  522.  
  523.     struct frag *cftt = (struct frag *) NULL; /* cur. frag to task */
  524.  
  525.     int cftd_togo;            /* len left to go to daemon */
  526.  
  527.     int frg_complete;        /* flag  if outgoing frag is complete */
  528.  
  529.  
  530. /* This is the relay process for MPI ports. 
  531.  *
  532.  * It should be called in pvm_mpp_beatask 
  533.  * after pvm_mpp_message_init, so none of the pre-allocated buffers have
  534.  * been allocated.
  535.  *
  536.  * It is basically a fragment forwarder
  537.  * pvmd -> task
  538.    - each fragment is read from the pvmd socket and forwarded to the
  539.      tasks.
  540.    task -> pvmd
  541.    - each fragment is read from a task and forwarded onto the pvmd socket.
  542.  
  543. */
  544.  
  545.     /* Initialization */
  546.  
  547.     mfunc = pvm_hostmsgfunc();
  548.  
  549.     nfds = dsock + 1;
  550.  
  551.     /* set up pre-allocated receive buffers to/from tasks */
  552.  
  553.     taskdirect = new_vdirectstruct(numnodes, NSBUFS, NRBUFS);
  554.  
  555.     taskfrags = init_recv_list(NSBUFS, PMTDBASE, MAXFRAGSIZE, 0, MPPANY,
  556.                     pvm_hostmsgfunc());
  557.  
  558.     pvmlogprintf("relay() numnodes is %d\n", numnodes);
  559.      /* initialize the packet numbering for packets to/from peers */    
  560.     for (i = 0; i < numnodes; i ++)
  561.     {
  562.         fill_directstruct (taskdirect + i, NRBUFS, i,
  563.             0, PMTDBASE, 0, MPPANY);
  564.         init_chunkostruct( (taskdirect+i)->ordering, NSBUFS);
  565.     }
  566.  
  567.     /* initialize the fragment queues */
  568.  
  569.      if( !(topvmdq = TALLOC(1, struct frag, "topvmdq")) 
  570.         || !(totaskq = TALLOC(1, struct frag, "totaskq")) )
  571.     {
  572.         pvmlogerror("relay() could not init frag queues\n");    
  573.         pvm_mpp_message_stop();
  574.         exit(PvmOutOfRes);
  575.     }    
  576.     BZERO((char *) topvmdq, sizeof (struct frag));
  577.     BZERO((char *) totaskq, sizeof (struct frag));
  578.  
  579.     topvmdq->fr_link = topvmdq->fr_rlink = topvmdq;
  580.     totaskq->fr_link = totaskq->fr_rlink = totaskq;
  581.  
  582.  
  583.     /* initialize the asynchronous send structures */    
  584.  
  585.     pvm_init_asynch_list(mppsendmids, mppoutchunks, NMPPSBUFMIDS);
  586.  
  587.     FD_ZERO(&wrk_rfds);        /* zero the file set descriptor for select */ 
  588.     FD_ZERO(&wrk_wfds);
  589.  
  590.     tout.tv_sec = tout.tv_usec = 0; /* don't block in select */
  591.     while (1) 
  592.     {            
  593.         /* Step 1) get any frags from tasks */
  594.  
  595.         while ( fp = (struct frag *) pvm_chunkReady(taskfrags, NRBUFS,
  596.                 mfunc, taskdirect, numnodes, &taskbuf, &readyFrags) )     
  597.         {
  598.             LISTPUTBEFORE(topvmdq, fp, fr_link, fr_rlink);
  599.             if (pvmdebmask & PDMNODE)
  600.             pvmlogprintf("Frag from task %d len %d \n", fp->fr_src, fp->fr_len);
  601.         }
  602.             
  603.  
  604.         /* If we don't have a frag to go to pvmd, check if one has been read */
  605.         if (!cftd && topvmdq->fr_link != topvmdq )
  606.         {
  607.             cftd = topvmdq->fr_link;
  608.             LISTDELETE(cftd, fr_link, fr_rlink);
  609.             cftd_togo = cftd->fr_len;
  610.             txcp = cftd->fr_dat;
  611.             if (pvmdebmask & PDMNODE)
  612.             pvmlogprintf("Frag to daemoen %d len %d \n", cftd->fr_src, cftd->fr_len);
  613.  
  614.         }
  615.  
  616.         /* There is a frag that needs to go to the daemon ? */
  617.         if (cftd)        
  618.         {
  619.             FD_SET(dsock, &wrk_wfds);
  620.             if (pvmdebmask & PDMNODE)
  621.             pvmlogprintf("Setting dsock in the write fds \n");
  622.         }
  623.  
  624.         FD_SET(dsock, &wrk_rfds); /* need to probe the socket for reading */
  625.  
  626.         rfds = wrk_rfds;
  627.         wfds = wrk_wfds;
  628.  
  629.         if (select(nfds, &rfds, &wfds, (fd_set*)0, &tout) == -1 
  630.         && errno != EINTR) {
  631.             pvmlogperror("relay() select");
  632.             pvm_exit();
  633.             pvm_mpp_message_stop();
  634.             exit(-1);
  635.         }
  636.  
  637.         /* we've got something read  from pvmd, if the frag is 
  638.             complete, then put it on the totaskq  */
  639.         if (FD_ISSET(dsock, &rfds)) 
  640.         {
  641.             frg_complete = 1; /* initially mark the frag as completely read */
  642.  
  643.             if (!frpvmd) {            /* allocate a data frag */
  644.                 frpvmd = fr_new(MAXFRAGSIZE);
  645.                 toread = TDFRAGHDR;
  646.             }
  647.  
  648.             n = read(dsock, frpvmd->fr_dat + frpvmd->fr_len, toread);
  649.  
  650.             if (n == -1 && errno != EWOULDBLOCK && errno != EINTR) {
  651.                 pvmlogperror("relay() read pvmd sock");
  652.                 pvm_mpp_message_stop();
  653.                 exit(12);
  654.             }
  655.             if (!n) {
  656.                 /* pvmlogerror("relay() read EOF on pvmd sock\n"); */
  657.                 pvm_mpp_message_stop();
  658.                 exit(13);
  659.             }
  660.  
  661.             if ((frpvmd->fr_len += n) < TDFRAGHDR) {
  662.                 toread -= n;
  663.                  frg_complete = 0;    
  664.             }
  665.  
  666.             
  667.             n = TDFRAGHDR + pvmget32(frpvmd->fr_dat + 8);   /* get the length */
  668.  
  669.             if (frpvmd->fr_len < n) {
  670.                 if (frpvmd->fr_max < n) {                    /* n > MAXFRAGSIZ */
  671.                     hdr = frpvmd;
  672.                     frpvmd = fr_new(n);
  673.                     BCOPY(hdr->fr_dat, frpvmd->fr_dat, hdr->fr_len);
  674.                     frpvmd->fr_len = hdr->fr_len;
  675.                     fr_unref(hdr);
  676.                 }
  677.                 toread = n - frpvmd->fr_len;
  678.                 frg_complete = 0;
  679.             }
  680.  
  681.             if (pvmdebmask & PDMNODE)
  682.             pvmlogprintf("reading from daemon len %d frg_complete %d \n", frpvmd->fr_len, frg_complete);
  683.             if (frg_complete)
  684.             {    
  685.                 dst = pvmget32(frpvmd->fr_dat);
  686.  
  687.                 frpvmd -> fr_src = dst; /* overload the src field with the dst */
  688.         
  689.                 LISTPUTAFTER(totaskq, frpvmd, fr_link, fr_rlink); /* queue it */
  690.  
  691.             if (pvmdebmask & PDMNODE)
  692.             pvmlogprintf("frg_complete dest %x len %d\n", frpvmd->fr_src, frpvmd->fr_len);
  693.                 frpvmd = (struct frag *) NULL;
  694.             }
  695.  
  696.         }
  697.  
  698.  
  699.         if (!cftt && totaskq -> fr_link != totaskq )
  700.         {
  701.             cftt = totaskq->fr_link;
  702.             LISTDELETE(cftt, fr_link, fr_rlink);
  703.             if (pvmdebmask & PDMNODE)
  704.             pvmlogprintf("New frag to task %x  len %d \n",
  705.                 cftt->fr_src, cftt->fr_len);
  706.         }
  707.  
  708.  
  709.         /* Let's see if we something to send to a task  */
  710.         if (cftt)
  711.         {
  712.             if (send_to_node(cftt,  taskdirect, numnodes) >= cftt->fr_len)
  713.                 cftt = (struct frag *) NULL;
  714.             else
  715.             /* if for some reason send_to_node could not send this frag,
  716.                the next time through the loop, the frag will be resent.
  717.             */;
  718.         }
  719.  
  720.         /* Now let's see if we have something to write to the pvmd */
  721.  
  722.         if ( cftd && FD_ISSET(dsock, &wfds))
  723.         {
  724.             if (pvmdebmask & PDMNODE)
  725.             pvmlogprintf("Writing packet to daemon %d\n", cftd->fr_len);
  726.             n = write(dsock, txcp, cftd_togo);
  727.  
  728.             if (n == -1 && errno != EWOULDBLOCK && errno != EINTR) {
  729.                 pvmlogperror("relay() write pvmd sock");
  730.                 pvm_mpp_message_stop();
  731.                 exit(14);
  732.             }
  733.             if (n > 0 && (cftd_togo -= n) > 0) 
  734.                 txcp += n;
  735.             if (!cftd_togo) {        /* entire message sent */
  736.                 FD_CLR(dsock, &wrk_wfds);
  737.                 fr_unref(cftd);
  738.                 cftd = (struct frag *) NULL;
  739.             }
  740.         }
  741.         else
  742.         {
  743.             if (cftd)
  744.             if (pvmdebmask & PDMNODE)
  745.                 pvmlogprintf("could not write to daemon \n");
  746.         }
  747.     }
  748. }
  749.  
  750. /* --------- send_to_node() ------------ */
  751. /* return the number of bytes sent. This posts sends asynchronously.
  752.    A fixed number of asynch sends are allowed to be outstanding.
  753.    If the relay asks for more than this number in sends. send_to_node
  754.    will return 0.
  755. */
  756.  
  757. send_to_node(fp,  taskdirect, numnodes )
  758. struct frag *fp; 
  759. MPP_DIRECTI_PTR taskdirect;
  760. int numnodes;
  761. {
  762.     int appid = 0;
  763.     int cc;
  764.     int idx;
  765.     int len;
  766.     int node;
  767.     int ptype;
  768.     int tag;
  769.     msgmid_t mid;
  770.  
  771.  
  772.     MPP_DIRECTI_PTR tdirect;
  773.  
  774.     MSGFUNC_PTR mfunc;
  775.  
  776.     mfunc = pvm_hostmsgfunc();
  777.  
  778.     if ( (idx = pvm_mpp_find_midx(mppsendmids, mppoutchunks, &lastindex, 
  779.                 NMPPSBUFMIDS, mfunc)) >= 0)
  780.     {
  781.         node = fp->fr_src & TIDNODE;
  782.         len = fp->fr_len;
  783.         ptype = (fp->fr_src & TIDPTYPE) >> (ffs(TIDPTYPE) - 1);
  784.  
  785.         tdirect = taskdirect + node;
  786.  
  787.         tag = tdirect->tagbase + tdirect->sseq;
  788.  
  789.         if ( ++(tdirect->sseq) >= tdirect->nbufs)
  790.             tdirect->sseq = 0;
  791.  
  792.             
  793.             if ( (cc  = (*mfunc->imsgsend)(appid, tag,
  794.                 fp->fr_dat, len, node, ptype, &mid)) < 0 )
  795.          {
  796.             pvmlogprintf("Relay -- async send failed!  (%d) \n", cc);
  797.         }
  798.         else /* send was ok. */
  799.         {
  800.             pvm_assign_mid(mppsendmids, mid, idx);
  801.             pvm_assign_chunk(mppoutchunks, (CHUNK_PTR) fp, idx);
  802.         }
  803.  
  804.         return (len);
  805.  
  806.     }
  807.     else
  808.         return 0;    /* could not send, no free low-level message ids */
  809. }
  810.     
  811.  
  812. /* --------- pvmhost() ------------ */
  813. /* We're the "host" process. Connect to pvmd. */
  814. int 
  815. pvmhost()
  816. {
  817.     char *p;
  818.     int dsock;                    /* pvmd socket */
  819.     struct sockaddr_in dsadr;    /* address of pvmd socket */
  820.     int n;
  821.     int i;
  822.     int cc;
  823.     int err;
  824.     
  825.     int pvminfo[SIZEHINFO];        /* ntask, hostpart, ptid, MTU, NDF */
  826.     char nullmsg[TDFRAGHDR+MSGHDRLEN];
  827.     int appid = 0;                /* Would be used for PGONPUMA */
  828.  
  829.     MSGFUNC_PTR mfunc;
  830.  
  831.     msgmid_t confmid;
  832.  
  833.     info_t confinfo;
  834.  
  835.     if (!(p = getenv("PVMSOCK"))) {
  836.         pvmlogerror("pvmhost() getenv() pvmd socket\n");
  837.         pvm_mpp_message_stop();
  838.         exit(2);
  839.     }
  840.     if ((dsock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
  841.         pvmlogperror("pvmhost() socket");
  842.         pvm_mpp_message_stop();
  843.         exit(3);
  844.     }
  845.     BZERO((char*)&dsadr, sizeof(dsadr));
  846.     hex_inadport(p, &dsadr);
  847.     dsadr.sin_family = AF_INET;
  848.     n = sizeof(dsadr);
  849.     while (connect(dsock, (struct sockaddr*)&dsadr, n) == -1)
  850.         if (errno != EINTR) {
  851.             pvmlogperror("pvmhost() connect");
  852.             pvm_mpp_message_stop();
  853.             exit(4);
  854.         }
  855. #ifndef NOSOCKOPT
  856.     n = 1;
  857.     if (setsockopt(dsock, IPPROTO_TCP, TCP_NODELAY, (char*)&n, sizeof(int))
  858.     == -1) {
  859.         pvmlogperror("pvmhost() setsockopt");
  860.         pvm_mpp_message_stop();
  861.         exit(5);
  862.     }
  863. #endif
  864.     if (!(p = getenv("PVMEPID"))) {
  865.         pvmlogerror("pvmhost() getenv() pid\n");
  866.         pvm_mpp_message_stop();
  867.         exit(6);
  868.     }
  869.     pvmmyupid = atoi(p);
  870.     BZERO(nullmsg, MAXHDR);
  871.     pvmput32(nullmsg, TIDPVMD);
  872.     pvmput32(nullmsg + 4, pvmmyupid);
  873.     pvmput32(nullmsg + 8, MSGHDRLEN);
  874.     pvmput32(nullmsg + 12, 0);            /* to keep putrify happy */
  875.     pvmput8(nullmsg + 12, FFSOM|FFEOM);
  876.     if (write(dsock, nullmsg, TDFRAGHDR+MSGHDRLEN) != TDFRAGHDR+MSGHDRLEN
  877.     || read(dsock, pvminfo, SIZEHINFO*sizeof(int)) != SIZEHINFO*sizeof(int)) {
  878.         pvmlogperror("pvmhost() write/read");
  879.         pvm_mpp_message_stop();
  880.         exit(8);
  881.     }
  882.     /* Multicast the configuration message */  
  883.     mfunc = pvm_hostmsgfunc();
  884.     pvmlogprintf("pvmhost() --  Starting configMessage send Loop %d\n",
  885.         pvmhostnode);
  886.     for (i = 0; i < pvmhostnode; i ++ )
  887.     {
  888.         if ( (*mfunc->imsgsend)(appid, PMTCONF,  (char *) pvminfo, 
  889.                 sizeof(pvminfo), i, PVMDPTYPE, &confmid) < 0) 
  890.         {
  891.             pvmlogperror("pvmhost() configuration message \n");
  892.             err = PvmDSysErr;
  893.             pvm_mpp_message_stop();
  894.             exit(9);
  895.         }
  896.         while (! ((*mfunc->msgdone)(appid, &confmid, &confinfo)));
  897.     }
  898.     pvmlogerror("pvmhost() -- Finished Message send Loop\n");
  899.     (void)relay(dsock, pvmhostnode + 1);
  900. }
  901.  
  902. #endif /*IMA_SP2MPI*/
  903.